/**
 *  deferredconsumerbase.cpp
 *
 *  Base class for the deferred consumer and the
 *  deferred get.
 *
 *  @copyright 2016 Copernica B.V.
 */

/**
 *  Dependencies
 */
#include "deferredconsumerbase.h"
#include "basicdeliverframe.h"
#include "basicgetokframe.h"
#include "basicheaderframe.h"
#include "bodyframe.h"

/**
 *  Start namespace
 */
namespace AMQP {

/**
 *  Process a delivery frame
 *
 *  @param  frame   The frame to process
 */
void DeferredConsumerBase::process(BasicDeliverFrame &frame)
{
    // retrieve the delivery tag and whether we were redelivered
    _deliveryTag = frame.deliveryTag();
    _redelivered = frame.redelivered();

    // anybody interested in the new message?
    if (_beginCallback)   _beginCallback();

    // do we have anybody interested in messages?
    if (_messageCallback) _message.construct(frame.exchange(), frame.routingKey());
}

/**
 *  Process a delivery frame from a get request
 *
 *  @param  frame   The frame to process
 */
void DeferredConsumerBase::process(BasicGetOKFrame &frame)
{
    // retrieve the delivery tag and whether we were redelivered
    _deliveryTag = frame.deliveryTag();
    _redelivered = frame.redelivered();

    // anybody interested in the new message?
    if (_beginCallback)   _beginCallback();

    // do we have anybody interested in messages?
    if (_messageCallback) _message.construct(frame.exchange(), frame.routingKey());
}

/**
 *  Indicate that a message was done
 */
void DeferredConsumerBase::complete()
{
    // make sure we stay in scope
    auto self = shared_from_this();
	auto &channel_val = _channel.lock();
    // also monitor the channel
	Monitor monitor{ channel_val };

    // do we have a message?
    if (_message)
    {
        // emit the message
        emit(std::move(*_message), _deliveryTag, _redelivered);

        // and destroy it
        _message.reset();
    }

    // do we have to inform anyone about completion?
    if (_completeCallback) _completeCallback(_deliveryTag, _redelivered);

    // do we still have a valid channel
    if (!monitor.valid()) return;

    // we are now done executing
	if (channel_val && channel_val.get())
		channel_val->complete();
}

/**
 *  End namespace
 */
}
